//! 简化的Flight SQL服务实现
//! 
//! 提供基础的Flight SQL协议支持

use std::pin::Pin;
use std::sync::Arc;
use futures::Stream;
use tonic::{Request, Response, Status};
use arrow_flight::{
    flight_service_server::FlightService, Action, ActionType, Criteria, Empty, FlightData,
    FlightDescriptor, FlightInfo, HandshakeRequest, HandshakeResponse, PutResult, SchemaResult,
    Ticket, Result as FlightResult,
};
use data_gateway_kernel::{DataGatewayKernel, Query, QueryType};
use data_gateway_kernel::query::{QueryOptions, QueryContext, QueryResultData};
use bytes::Bytes;
use serde_json::Value;
use arrow::array::{StringArray, RecordBatch};
use arrow::datatypes::{Schema, Field, DataType};
use std::collections::HashMap;
use tokio::sync::RwLock;

/// 准备语句信息
#[derive(Debug, Clone)]
pub struct PreparedStatement {
    /// 语句ID
    pub id: String,
    /// SQL语句
    pub sql: String,
    /// 参数Schema
    pub parameter_schema: Option<Schema>,
    /// 结果Schema
    pub result_schema: Option<Schema>,
    /// 创建时间
    pub created_at: std::time::SystemTime,
}

/// 事务状态
#[derive(Debug, Clone, PartialEq)]
pub enum TransactionStatus {
    /// 活跃状态
    Active,
    /// 已提交
    Committed,
    /// 已回滚
    RolledBack,
}

/// 事务信息
#[derive(Debug, Clone)]
pub struct Transaction {
    /// 事务ID
    pub id: String,
    /// 事务状态
    pub status: TransactionStatus,
    /// 创建时间
    pub created_at: std::time::SystemTime,
    /// 最后活动时间
    pub last_activity: std::time::SystemTime,
    /// 事务中执行的SQL语句
    pub statements: Vec<String>,
}

/// 简化的Flight SQL服务
#[derive(Clone)]
pub struct SimpleFlightSqlService {
    /// 内核引用
    kernel: Arc<DataGatewayKernel>,
    /// 准备语句存储
    prepared_statements: Arc<RwLock<HashMap<String, PreparedStatement>>>,
    /// 事务存储
    transactions: Arc<RwLock<HashMap<String, Transaction>>>,
}

impl SimpleFlightSqlService {
    /// 创建新的Flight SQL服务
    pub fn new(kernel: Arc<DataGatewayKernel>) -> Self {
        Self {
            kernel,
            prepared_statements: Arc::new(RwLock::new(HashMap::new())),
            transactions: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    /// 创建准备语句
    pub async fn create_prepared_statement(&self, sql: &str) -> Result<String, Status> {
        // 生成唯一ID
        let statement_id = uuid::Uuid::new_v4().to_string();

        // 创建准备语句对象
        let prepared_stmt = PreparedStatement {
            id: statement_id.clone(),
            sql: sql.to_string(),
            parameter_schema: None, // 简化实现，暂不支持参数
            result_schema: None,    // 延迟获取结果Schema
            created_at: std::time::SystemTime::now(),
        };

        // 存储准备语句
        let mut statements = self.prepared_statements.write().await;
        statements.insert(statement_id.clone(), prepared_stmt);

        Ok(statement_id)
    }

    /// 执行准备语句
    pub async fn execute_prepared_statement(&self, statement_id: &str, parameters: HashMap<String, Value>) -> Result<RecordBatch, Status> {
        // 获取准备语句
        let statements = self.prepared_statements.read().await;
        let prepared_stmt = statements.get(statement_id)
            .ok_or_else(|| Status::not_found(format!("准备语句未找到: {}", statement_id)))?;

        // 执行SQL查询（简化实现，暂不处理参数替换）
        self.execute_sql_query(&prepared_stmt.sql).await
    }

    /// 关闭准备语句
    pub async fn close_prepared_statement(&self, statement_id: &str) -> Result<(), Status> {
        let mut statements = self.prepared_statements.write().await;
        statements.remove(statement_id)
            .ok_or_else(|| Status::not_found(format!("准备语句未找到: {}", statement_id)))?;
        Ok(())
    }

    /// 获取准备语句信息
    pub async fn get_prepared_statement(&self, statement_id: &str) -> Result<PreparedStatement, Status> {
        let statements = self.prepared_statements.read().await;
        statements.get(statement_id)
            .cloned()
            .ok_or_else(|| Status::not_found(format!("准备语句未找到: {}", statement_id)))
    }

    /// 开始事务
    pub async fn begin_transaction(&self) -> Result<String, Status> {
        let transaction_id = uuid::Uuid::new_v4().to_string();
        let now = std::time::SystemTime::now();

        let transaction = Transaction {
            id: transaction_id.clone(),
            status: TransactionStatus::Active,
            created_at: now,
            last_activity: now,
            statements: Vec::new(),
        };

        let mut transactions = self.transactions.write().await;
        transactions.insert(transaction_id.clone(), transaction);

        Ok(transaction_id)
    }

    /// 提交事务
    pub async fn commit_transaction(&self, transaction_id: &str) -> Result<(), Status> {
        let mut transactions = self.transactions.write().await;
        let transaction = transactions.get_mut(transaction_id)
            .ok_or_else(|| Status::not_found(format!("事务未找到: {}", transaction_id)))?;

        if transaction.status != TransactionStatus::Active {
            return Err(Status::failed_precondition(format!("事务状态无效: {:?}", transaction.status)));
        }

        // 简化实现：直接标记为已提交
        transaction.status = TransactionStatus::Committed;
        transaction.last_activity = std::time::SystemTime::now();

        Ok(())
    }

    /// 回滚事务
    pub async fn rollback_transaction(&self, transaction_id: &str) -> Result<(), Status> {
        let mut transactions = self.transactions.write().await;
        let transaction = transactions.get_mut(transaction_id)
            .ok_or_else(|| Status::not_found(format!("事务未找到: {}", transaction_id)))?;

        if transaction.status != TransactionStatus::Active {
            return Err(Status::failed_precondition(format!("事务状态无效: {:?}", transaction.status)));
        }

        // 简化实现：直接标记为已回滚
        transaction.status = TransactionStatus::RolledBack;
        transaction.last_activity = std::time::SystemTime::now();

        Ok(())
    }

    /// 获取事务信息
    pub async fn get_transaction(&self, transaction_id: &str) -> Result<Transaction, Status> {
        let transactions = self.transactions.read().await;
        transactions.get(transaction_id)
            .cloned()
            .ok_or_else(|| Status::not_found(format!("事务未找到: {}", transaction_id)))
    }

    /// 清理过期事务（超过1小时的非活跃事务）
    pub async fn cleanup_expired_transactions(&self) -> Result<usize, Status> {
        let mut transactions = self.transactions.write().await;
        let now = std::time::SystemTime::now();
        let timeout = std::time::Duration::from_secs(3600); // 1小时

        let mut expired_ids = Vec::new();
        for (id, transaction) in transactions.iter() {
            if let Ok(elapsed) = now.duration_since(transaction.last_activity) {
                if elapsed > timeout && transaction.status == TransactionStatus::Active {
                    expired_ids.push(id.clone());
                }
            }
        }

        let count = expired_ids.len();
        for id in expired_ids {
            if let Some(mut transaction) = transactions.get_mut(&id) {
                transaction.status = TransactionStatus::RolledBack;
            }
        }

        Ok(count)
    }

    /// 执行SQL查询并返回Arrow RecordBatch
    async fn execute_sql_query(&self, sql: &str) -> Result<RecordBatch, Status> {
        // 创建查询对象
        let query = Query {
            id: uuid::Uuid::new_v4().to_string(),
            query_type: QueryType::Sql,
            content: sql.to_string(),
            parameters: HashMap::new(),
            options: QueryOptions::default(),
            context: QueryContext::default(),
        };

        // 执行查询
        match self.kernel.handle_query(query).await {
            Ok(result) => {
                // 将查询结果转换为Arrow RecordBatch
                self.convert_query_result_to_record_batch(result.data)
            }
            Err(e) => Err(Status::internal(format!("查询执行失败: {}", e))),
        }
    }

    /// 将查询结果转换为Arrow RecordBatch
    fn convert_query_result_to_record_batch(&self, data: QueryResultData) -> Result<RecordBatch, Status> {
        match data {
            QueryResultData::Json(json_data) => self.convert_json_to_record_batch(json_data),
            QueryResultData::Arrow(batches) => {
                if batches.is_empty() {
                    // 返回空的RecordBatch
                    let schema = Schema::new(vec![
                        Field::new("message", DataType::Utf8, false),
                    ]);
                    let message_array = StringArray::from(vec!["No data"]);
                    RecordBatch::try_new(Arc::new(schema), vec![Arc::new(message_array)])
                        .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
                } else {
                    // 返回第一个RecordBatch
                    Ok(batches[0].clone())
                }
            }
            QueryResultData::Raw(text) => {
                let schema = Schema::new(vec![
                    Field::new("result", DataType::Utf8, false),
                ]);
                let result_array = StringArray::from(vec![text]);
                RecordBatch::try_new(Arc::new(schema), vec![Arc::new(result_array)])
                    .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
            }
            QueryResultData::Empty => {
                let schema = Schema::new(vec![
                    Field::new("message", DataType::Utf8, false),
                ]);
                let message_array = StringArray::from(vec!["Empty result"]);
                RecordBatch::try_new(Arc::new(schema), vec![Arc::new(message_array)])
                    .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
            }
        }
    }

    /// 将JSON结果转换为Arrow RecordBatch
    fn convert_json_to_record_batch(&self, data: Value) -> Result<RecordBatch, Status> {
        match data {
            Value::Array(rows) => {
                if rows.is_empty() {
                    // 返回空的RecordBatch
                    let schema = Schema::new(vec![
                        Field::new("message", DataType::Utf8, false),
                    ]);
                    let message_array = StringArray::from(vec!["No data"]);
                    RecordBatch::try_new(Arc::new(schema), vec![Arc::new(message_array)])
                        .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
                } else {
                    // 简化实现：假设所有行都有相同的结构
                    if let Some(Value::Object(first_row)) = rows.first() {
                        let mut fields = Vec::new();
                        let mut arrays: Vec<Arc<dyn arrow::array::Array>> = Vec::new();

                        // 为每个字段创建数组
                        for (key, _) in first_row {
                            fields.push(Field::new(key, DataType::Utf8, true));

                            let values: Vec<Option<String>> = rows.iter().map(|row| {
                                if let Value::Object(obj) = row {
                                    obj.get(key).map(|v| v.to_string())
                                } else {
                                    None
                                }
                            }).collect();

                            arrays.push(Arc::new(StringArray::from(values)));
                        }

                        let schema = Schema::new(fields);
                        RecordBatch::try_new(Arc::new(schema), arrays)
                            .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
                    } else {
                        Err(Status::internal("无效的数据格式"))
                    }
                }
            }
            Value::Object(obj) => {
                // 单个对象，转换为单行RecordBatch
                let mut fields = Vec::new();
                let mut arrays: Vec<Arc<dyn arrow::array::Array>> = Vec::new();

                for (key, value) in obj {
                    fields.push(Field::new(&key, DataType::Utf8, true));
                    arrays.push(Arc::new(StringArray::from(vec![Some(value.to_string())])));
                }

                let schema = Schema::new(fields);
                RecordBatch::try_new(Arc::new(schema), arrays)
                    .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
            }
            _ => {
                // 其他类型，转换为简单的消息
                let schema = Schema::new(vec![
                    Field::new("result", DataType::Utf8, false),
                ]);
                let result_array = StringArray::from(vec![data.to_string()]);
                RecordBatch::try_new(Arc::new(schema), vec![Arc::new(result_array)])
                    .map_err(|e| Status::internal(format!("创建RecordBatch失败: {}", e)))
            }
        }
    }

    /// 将RecordBatch转换为FlightData
    fn record_batch_to_flight_data(&self, record_batch: RecordBatch) -> Result<FlightData, Status> {
        use arrow::ipc::writer::IpcWriteOptions;
        use arrow::ipc::writer::StreamWriter;
        use std::io::Cursor;

        // 创建内存缓冲区
        let mut buffer = Vec::new();
        let cursor = Cursor::new(&mut buffer);

        // 使用Arrow IPC格式序列化RecordBatch
        let options = IpcWriteOptions::default();
        let mut writer = StreamWriter::try_new_with_options(cursor, &record_batch.schema(), options)
            .map_err(|e| Status::internal(format!("创建IPC writer失败: {}", e)))?;

        writer.write(&record_batch)
            .map_err(|e| Status::internal(format!("写入RecordBatch失败: {}", e)))?;

        writer.finish()
            .map_err(|e| Status::internal(format!("完成写入失败: {}", e)))?;

        Ok(FlightData {
            data_header: Bytes::new(),
            data_body: Bytes::from(buffer),
            app_metadata: Bytes::new(),
            flight_descriptor: None,
        })
    }

    /// 序列化Arrow Schema
    fn serialize_schema(&self, schema: &Schema) -> Result<Bytes, Status> {
        use arrow::ipc::writer::IpcWriteOptions;
        use arrow::ipc::convert::schema_to_fb_offset;
        use flatbuffers::FlatBufferBuilder;

        let mut fbb = FlatBufferBuilder::new();
        let schema_offset = schema_to_fb_offset(&mut fbb, schema);
        fbb.finish(schema_offset, None);

        Ok(Bytes::from(fbb.finished_data().to_vec()))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use data_gateway_kernel::init_kernel;

    #[tokio::test]
    async fn test_flight_sql_service_creation() {
        let kernel = init_kernel().await.unwrap();
        let service = SimpleFlightSqlService::new(Arc::new(kernel));

        // 验证服务创建成功
        assert!(!service.kernel.get_stats().await.unwrap().state.is_empty());
    }

    #[tokio::test]
    async fn test_convert_json_to_record_batch() {
        let kernel = init_kernel().await.unwrap();
        let service = SimpleFlightSqlService::new(Arc::new(kernel));

        // 测试JSON数组转换
        let json_data = serde_json::json!([
            {"id": 1, "name": "Alice"},
            {"id": 2, "name": "Bob"}
        ]);

        let result = service.convert_json_to_record_batch(json_data);
        assert!(result.is_ok());

        let record_batch = result.unwrap();
        assert_eq!(record_batch.num_rows(), 2);
        assert_eq!(record_batch.num_columns(), 2);
    }

    #[tokio::test]
    async fn test_convert_query_result_to_record_batch() {
        let kernel = init_kernel().await.unwrap();
        let service = SimpleFlightSqlService::new(Arc::new(kernel));

        // 测试JSON结果转换
        let json_data = serde_json::json!({"message": "success"});
        let query_result = QueryResultData::Json(json_data);

        let result = service.convert_query_result_to_record_batch(query_result);
        assert!(result.is_ok());

        let record_batch = result.unwrap();
        assert_eq!(record_batch.num_rows(), 1);
        assert_eq!(record_batch.num_columns(), 1);
    }

    #[tokio::test]
    async fn test_serialize_schema() {
        let kernel = init_kernel().await.unwrap();
        let service = SimpleFlightSqlService::new(Arc::new(kernel));

        let schema = Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("name", DataType::Utf8, true),
        ]);

        let result = service.serialize_schema(&schema);
        assert!(result.is_ok());

        let serialized = result.unwrap();
        assert!(!serialized.is_empty());
    }
}

#[tonic::async_trait]
impl FlightService for SimpleFlightSqlService {
    type HandshakeStream = Pin<Box<dyn Stream<Item = std::result::Result<HandshakeResponse, Status>> + Send>>;
    type ListFlightsStream = Pin<Box<dyn Stream<Item = std::result::Result<FlightInfo, Status>> + Send>>;
    type DoGetStream = Pin<Box<dyn Stream<Item = std::result::Result<FlightData, Status>> + Send>>;
    type DoPutStream = Pin<Box<dyn Stream<Item = std::result::Result<PutResult, Status>> + Send>>;
    type DoActionStream = Pin<Box<dyn Stream<Item = std::result::Result<FlightResult, Status>> + Send>>;
    type ListActionsStream = Pin<Box<dyn Stream<Item = std::result::Result<ActionType, Status>> + Send>>;
    type DoExchangeStream = Pin<Box<dyn Stream<Item = std::result::Result<FlightData, Status>> + Send>>;

    async fn handshake(
        &self,
        _request: Request<tonic::Streaming<HandshakeRequest>>,
    ) -> std::result::Result<Response<Self::HandshakeStream>, Status> {
        let output = futures::stream::once(async {
            Ok(HandshakeResponse {
                protocol_version: 0,
                payload: Bytes::from("Flight SQL Server Ready"),
            })
        });
        Ok(Response::new(Box::pin(output)))
    }

    async fn list_flights(
        &self,
        _request: Request<Criteria>,
    ) -> std::result::Result<Response<Self::ListFlightsStream>, Status> {
        let output = futures::stream::empty();
        Ok(Response::new(Box::pin(output)))
    }

    async fn get_flight_info(
        &self,
        request: Request<FlightDescriptor>,
    ) -> std::result::Result<Response<FlightInfo>, Status> {
        let descriptor = request.into_inner();

        // 从descriptor中提取SQL查询
        let sql = if !descriptor.cmd.is_empty() {
            String::from_utf8(descriptor.cmd.to_vec())
                .map_err(|e| Status::invalid_argument(format!("无效的命令格式: {}", e)))?
        } else {
            return Err(Status::invalid_argument("缺少SQL命令"));
        };

        // 执行查询以获取schema信息
        let record_batch = self.execute_sql_query(&sql).await?;

        // 序列化schema
        let schema_bytes = self.serialize_schema(&record_batch.schema())?;

        // 创建ticket用于后续的do_get调用
        let ticket = arrow_flight::Ticket {
            ticket: Bytes::from(sql.clone().into_bytes()),
        };

        // 创建endpoint
        let endpoint = arrow_flight::FlightEndpoint {
            ticket: Some(ticket),
            location: vec![], // 本地服务，不需要location
            expiration_time: None,
            app_metadata: Bytes::new(),
        };

        let flight_info = FlightInfo {
            schema: schema_bytes,
            flight_descriptor: Some(descriptor),
            endpoint: vec![endpoint],
            total_records: record_batch.num_rows() as i64,
            total_bytes: -1, // 暂时不计算字节数
            ordered: false,
            app_metadata: Bytes::new(),
        };

        Ok(Response::new(flight_info))
    }

    async fn poll_flight_info(
        &self,
        _request: Request<FlightDescriptor>,
    ) -> std::result::Result<Response<arrow_flight::PollInfo>, Status> {
        let poll_info = arrow_flight::PollInfo {
            info: None,
            flight_descriptor: None,
            progress: None,
            expiration_time: None,
        };
        Ok(Response::new(poll_info))
    }

    async fn get_schema(
        &self,
        _request: Request<FlightDescriptor>,
    ) -> std::result::Result<Response<SchemaResult>, Status> {
        let schema_result = SchemaResult {
            schema: Bytes::new(),
        };
        Ok(Response::new(schema_result))
    }

    async fn do_get(
        &self,
        request: Request<Ticket>,
    ) -> std::result::Result<Response<Self::DoGetStream>, Status> {
        let ticket = request.into_inner();

        // 从ticket中解析SQL查询
        let sql = String::from_utf8(ticket.ticket.to_vec())
            .map_err(|e| Status::invalid_argument(format!("无效的ticket格式: {}", e)))?;

        // 执行SQL查询
        let record_batch = self.execute_sql_query(&sql).await?;

        // 将RecordBatch转换为FlightData
        let flight_data = self.record_batch_to_flight_data(record_batch)?;

        let output = futures::stream::once(async move { Ok(flight_data) });
        Ok(Response::new(Box::pin(output)))
    }

    async fn do_put(
        &self,
        _request: Request<tonic::Streaming<FlightData>>,
    ) -> std::result::Result<Response<Self::DoPutStream>, Status> {
        let output = futures::stream::once(async {
            Ok(PutResult {
                app_metadata: Bytes::new(),
            })
        });
        Ok(Response::new(Box::pin(output)))
    }

    async fn do_action(
        &self,
        _request: Request<Action>,
    ) -> std::result::Result<Response<Self::DoActionStream>, Status> {
        let output = futures::stream::once(async {
            Ok(FlightResult {
                body: Bytes::from("OK"),
            })
        });
        Ok(Response::new(Box::pin(output)))
    }

    async fn list_actions(
        &self,
        _request: Request<Empty>,
    ) -> std::result::Result<Response<Self::ListActionsStream>, Status> {
        let output = futures::stream::empty();
        Ok(Response::new(Box::pin(output)))
    }

    async fn do_exchange(
        &self,
        _request: Request<tonic::Streaming<FlightData>>,
    ) -> std::result::Result<Response<Self::DoExchangeStream>, Status> {
        let output = futures::stream::empty();
        Ok(Response::new(Box::pin(output)))
    }
}

#[cfg(test)]
mod advanced_tests {
    use super::*;
    use data_gateway_kernel::DataGatewayKernel;

    async fn create_test_service() -> SimpleFlightSqlService {
        let kernel = Arc::new(DataGatewayKernel::new().await.unwrap());
        SimpleFlightSqlService::new(kernel)
    }

    #[tokio::test]
    async fn test_prepared_statement_lifecycle() {
        let service = create_test_service().await;

        // 创建准备语句
        let sql = "SELECT * FROM users WHERE id = ?";
        let stmt_id = service.create_prepared_statement(sql).await.unwrap();
        assert!(!stmt_id.is_empty());

        // 获取准备语句信息
        let stmt = service.get_prepared_statement(&stmt_id).await.unwrap();
        assert_eq!(stmt.sql, sql);
        assert_eq!(stmt.id, stmt_id);

        // 关闭准备语句
        service.close_prepared_statement(&stmt_id).await.unwrap();

        // 验证语句已被删除
        assert!(service.get_prepared_statement(&stmt_id).await.is_err());
    }

    #[tokio::test]
    async fn test_prepared_statement_execution() {
        let service = create_test_service().await;

        // 创建准备语句
        let sql = "SELECT 1 as test_column";
        let stmt_id = service.create_prepared_statement(sql).await.unwrap();

        // 执行准备语句
        let parameters = HashMap::new();
        let result = service.execute_prepared_statement(&stmt_id, parameters).await;

        // 简化实现可能会失败，但不应该panic
        match result {
            Ok(_) => println!("准备语句执行成功"),
            Err(e) => println!("准备语句执行失败: {}", e),
        }

        // 清理
        service.close_prepared_statement(&stmt_id).await.unwrap();
    }

    #[tokio::test]
    async fn test_transaction_lifecycle() {
        let service = create_test_service().await;

        // 开始事务
        let tx_id = service.begin_transaction().await.unwrap();
        assert!(!tx_id.is_empty());

        // 获取事务信息
        let tx = service.get_transaction(&tx_id).await.unwrap();
        assert_eq!(tx.id, tx_id);
        assert_eq!(tx.status, TransactionStatus::Active);

        // 提交事务
        service.commit_transaction(&tx_id).await.unwrap();

        // 验证事务状态
        let tx = service.get_transaction(&tx_id).await.unwrap();
        assert_eq!(tx.status, TransactionStatus::Committed);
    }

    #[tokio::test]
    async fn test_transaction_rollback() {
        let service = create_test_service().await;

        // 开始事务
        let tx_id = service.begin_transaction().await.unwrap();

        // 回滚事务
        service.rollback_transaction(&tx_id).await.unwrap();

        // 验证事务状态
        let tx = service.get_transaction(&tx_id).await.unwrap();
        assert_eq!(tx.status, TransactionStatus::RolledBack);
    }

    #[tokio::test]
    async fn test_transaction_invalid_operations() {
        let service = create_test_service().await;

        // 开始事务
        let tx_id = service.begin_transaction().await.unwrap();

        // 提交事务
        service.commit_transaction(&tx_id).await.unwrap();

        // 尝试再次提交应该失败
        assert!(service.commit_transaction(&tx_id).await.is_err());

        // 尝试回滚已提交的事务应该失败
        assert!(service.rollback_transaction(&tx_id).await.is_err());
    }

    #[tokio::test]
    async fn test_cleanup_expired_transactions() {
        let service = create_test_service().await;

        // 创建一个事务
        let tx_id = service.begin_transaction().await.unwrap();

        // 手动修改事务的最后活动时间为2小时前
        {
            let mut transactions = service.transactions.write().await;
            if let Some(tx) = transactions.get_mut(&tx_id) {
                tx.last_activity = std::time::SystemTime::now() - std::time::Duration::from_secs(7200);
            }
        }

        // 清理过期事务
        let cleaned_count = service.cleanup_expired_transactions().await.unwrap();
        assert_eq!(cleaned_count, 1);

        // 验证事务状态已变为回滚
        let tx = service.get_transaction(&tx_id).await.unwrap();
        assert_eq!(tx.status, TransactionStatus::RolledBack);
    }

    #[tokio::test]
    async fn test_nonexistent_prepared_statement() {
        let service = create_test_service().await;

        let fake_id = "nonexistent-id";

        // 获取不存在的准备语句应该失败
        assert!(service.get_prepared_statement(fake_id).await.is_err());

        // 执行不存在的准备语句应该失败
        let parameters = HashMap::new();
        assert!(service.execute_prepared_statement(fake_id, parameters).await.is_err());

        // 关闭不存在的准备语句应该失败
        assert!(service.close_prepared_statement(fake_id).await.is_err());
    }

    #[tokio::test]
    async fn test_nonexistent_transaction() {
        let service = create_test_service().await;

        let fake_id = "nonexistent-tx-id";

        // 获取不存在的事务应该失败
        assert!(service.get_transaction(fake_id).await.is_err());

        // 提交不存在的事务应该失败
        assert!(service.commit_transaction(fake_id).await.is_err());

        // 回滚不存在的事务应该失败
        assert!(service.rollback_transaction(fake_id).await.is_err());
    }
}
